const kafka = require('kafka-node');
const ConsumerGroup = kafka.ConsumerGroup;
const topic = process.env.NODE_ENV === 'production' ? 'buried_point' : 'log_test'

const options = {
  kafkaHost: '172.16.3.62:9092,172.16.3.65:9092,172.16.3.67:9092',
  batch: undefined, // put client batch settings if you need them
  groupId: 'node-logdebug',//consumer group id, default `kafka-node-group`
  sessionTimeout: 15000,
  // An array of partition assignment protocols ordered by preference.
  // 'roundrobin' or 'range' string for built ins (see below to pass in custom assignment protocol)
  protocol: ['roundrobin'],
  encoding: 'utf8', // default is utf8, use 'buffer' for binary data
  // Offsets to use for new groups other options could be 'earliest' or 'none' (none will emit an error if no offsets were saved)
  // equivalent to Java client's auto.offset.reset
  fromOffset: 'latest', // default
  commitOffsetsOnFirstJoin: true, // on the very first time this consumer group subscribes to a topic, record the offset returned in fromOffset (latest/earliest)
  // how to recover from OutOfRangeOffset error (where save offset is past server retention) accepts same value as fromOffset
  outOfRangeOffset: 'earliest', // default
  // Callback to allow consumers with autoCommit false a chance to commit before a rebalance finishes
  // isAlreadyMember will be false on the first connection, and true on rebalances triggered after that
  onRebalance: (isAlreadyMember, callback) => { callback(); } // or null
};
const consumer = new ConsumerGroup(options, topic);

const moonids = ["uhbw5w5bw", "z4vqsqclx", "anywd3b1d"];

consumer.on('message', function (message) {
  try {
    const value = JSON.parse(message.value)
    if (moonids.indexOf(value.moonid) > -1) {
      if (value.t == 'js') {
        
      } else if (value.t == 'oper') {

      }
    }
  } catch (e) { }
  console.log(message);
});